package com.upcoming.data.storm.util;

import org.apache.kafka.common.serialization.Serializer;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.util.Map;

public class EventEncoder implements Serializer<Event> {
	@Override
	public void configure(Map<String, ?> map, boolean b) {

	}

	@Override
	public byte[] serialize(String s, Event event) {
		return ObjectToBytes(event);
	}

	@Override
	public void close() {

	}


	/**
	 * 对象转字节数组
	 *
	 * @param obj
	 * @return
	 */
	private static byte[] ObjectToBytes(Object obj) {
		byte[] bytes = null;
		ByteArrayOutputStream bo = null;
		ObjectOutputStream oo = null;
		try {
			bo = new ByteArrayOutputStream();
			oo = new ObjectOutputStream(bo);
			oo.writeObject(obj);
			bytes = bo.toByteArray();

		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			try {
				if (bo != null) {
					bo.close();
				}
				if (oo != null) {
					oo.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return bytes;
	}



}
